package org.springblade.core.etl.service;

import cn.hutool.core.util.IdUtil;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springblade.core.etl.config.EtlSourceConfig;
import org.springblade.core.etl.dao.ClientHelper;
import org.springblade.core.etl.model.EtlRunVO;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class EtlJobManager {
    /**
     * 作业
     */
    private static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(80);

    /**
     * 作业缓存
     */
    private static Map<String, EtlJob> jobCache = new HashMap<>();

    /**
     * 添加作业
     *
     * @param jobCode      编码
     * @param job          作业对象
     * @param config       配置对象
     * @param defaultDelay 单位是秒
     */
    public static void addJob(final String jobCode, final EtlJob job, final EtlSourceConfig config, final Integer defaultDelay) {
        if (jobCache.containsKey(jobCode)) {
            throw new RuntimeException("作业编码" + jobCode + "重复");
        }
        jobCache.put(jobCode, job);
        int first = Double.valueOf(Math.random() * 100).intValue();
        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                runJob(jobCode, config);
            }
        }, first, defaultDelay, TimeUnit.SECONDS);

    }


    /**
     * 添加作业
     *
     * @param jobCode 编码
     * @param config  配置对象
     */
    public static void runJob(final String jobCode, final EtlSourceConfig config) {
        //作业名称
        EtlJob job = jobCache.get(jobCode);
        if (job == null) {
            return;
        }
        if (!"Y".equals(config.getEnable())) {
            log.error(jobCode + "作业执行,跳过");
            return;
        }
        String traceId = IdUtil.fastSimpleUUID();
        MDC.put("traceId", IdUtil.fastSimpleUUID());
        Long start = System.currentTimeMillis();
        log.info(jobCode + "作业执行开始》》");
        try {
            ClientHelper.setDefault(jobCode);
            EtlRunVO run = new EtlRunVO();
            run.setTraceId(traceId);
            job.process(run);
        } catch (Exception ex) {
            log.error(jobCode + "作业执行异常" + ex.getMessage(), ex);
        } finally {
            log.info(jobCode + "作业执行结束《《   耗时：{}ms", (System.currentTimeMillis() - start));
        }

    }
}
